package test.kafka.KafkaPushDemo;

import java.util.Properties;

import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

/**
 * Hello world!
 *
 */
public class App {
    public static void main( String[] args ){
    	 Properties props = new Properties();
    	 String kafkaServer=PropertiesUtil.getStringPro("kafka.server");
    	 String total=PropertiesUtil.getStringPro("total");
    	 boolean isssl=PropertiesUtil.getbooleanPro("isssl");
    	 int totalTiaoshu=0;
    	 if(StringUtils.isBlank(total)){
    		 totalTiaoshu=100;
    	 }else{
    		 totalTiaoshu=Integer.parseInt(total);
    	 }
         props.put("bootstrap.servers",kafkaServer);
         props.put("enable.auto.commit", "true");
         props.put("auto.commit.interval.ms", "1000");
         props.put("session.timeout.ms", "30000");
         props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
         props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
         if(isssl){
        	 props.put("security.protocol", "SSL");
             props.put("ssl.truststore.location", "E:\\client\\client.truststore.jks");
             props.put("ssl.truststore.password", "tykj1234");
             props.put("ssl.keystore.location", "E:\\client\\client1.keystore.jks");
             props.put("ssl.keystore.password", "tykj1234");
             props.put("ssl.key.password", "tykj1234");
         }
         props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "1000");
         KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);
         System.out.println("-----------------------------kafka建立连接中------------------------------------------");
         producer.send(new ProducerRecord<>("TY_PS_KTP_CCITest", "device",""));
         System.out.println("-----------------------------数据发送中("+totalTiaoshu+"条）------------------------------------------");
         for(int i=0;i<totalTiaoshu;i++){
        	 String jsonStr="你们好！！";
        	 producer.send(new ProducerRecord<>("TY_PS_KTP_CCITest", "device",jsonStr));
         }
         System.out.println("-----------------------------success------------------------------------------");
         producer.flush();
    }
}
